package com.fyy.kafuka.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * @Author fyy
 * @Description
 * @Date 2020/11/6 16:07
 */
public class CountInterceptor implements ProducerInterceptor<String,String> {

    private int success;
    private int error;

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
        return producerRecord;
    }
    //在发送之后调用
    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
        if(recordMetadata != null){
            success++;
        }else{
            error++;
        }
    }

    @Override
    public void close() {
        System.out.println("成功发送："+success+"条记录");
        System.out.println("失败发送："+error+"条记录");
    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}
